<?php
/*
 *  This file is part of Urd.
 *
 *  Urd is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *  Urd is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program. See the file "COPYING". If it does not
 *  exist, see <http://www.gnu.org/licenses/>.
 *
 * $LastChangedDate: 2008-09-13 00:38:08 +0200 (Sat, 13 Sep 2008) $
 * $Rev: 225 $
 * $Author: gavinspearhead $
 * $Id: urdd_usenet_servers.php 225 2008-09-12 22:38:08Z gavinspearhead $
 */

if (!defined('ORIGINAL_PAGE')) die('This file cannot be accessed directly.');


class server_data { // lots of cleaning up to do
	private $servers; // these are all the news servers that we know of, each has its own thread_list
	private $schedule; // here we store all the scheduled jobs
	private $queue; // this is were all the tasks are put before they can run
	private $threads; // this is for non-nntp threads; other threads run in the usenet_servers' threads 
	private $max_total_nntp_threads; // total number of threads that can run on all newsgroups together
	private $free_nntp_slots; // available threads for treads on newgroups
	private $max_total_threads; // total number of threads that can run all together (always at least one more than the nntp threads, so we can do unrar and stuff
	private $free_total_slots; // total number of threads that are available
	private $conn_check_time; // time after which connections must be reset;
	private $max_db_intensive_threads; // max number of threads with db intensive set that can run
	private $free_db_intensive_slots; // available number of threads with db intensive set that can run
	// generic

	function print_size() 
	{ 
		echo_debug("{$this->max_total_nntp_threads} {$this->free_nntp_slots} {$this->max_total_threads} {$this->free_total_slots} {$this->max_db_intensive_threads} {$this->free_db_intensive_slots}", DEBUG_SERVER);
		$this->servers->print_size();
	}
	function __construct($queue_size, $nntp_threads, $total_threads, $db_intensive_threads)
	{
		assert(is_numeric($queue_size) && is_numeric($nntp_threads) && is_numeric($total_threads) && is_numeric($db_intensive_threads));
		$this->threads = new thread_list();
		$this->servers = new usenet_servers();
		$this->queue = new queue($queue_size);
		$this->schedule = new schedule();
		if ($total_threads < $nntp_threads)
			$total_threads = $nntp_threads + 1; // the default is we have always one slot available for other things
		$this->free_nntp_slots = $this->max_total_nntp_threads = (int) $nntp_threads;
		$this->free_total_slots = $this->max_total_threads = (int) $total_threads;
		$this->free_db_intensive_slots = $this->max_db_intensive_threads = (int) $db_intensive_threads;
		$this->conn_check_time = (int) 0;
	}
	function check_conn_time()
	{
		if (($this->conn_check_time > 0) && (time() > $this->conn_check_time)) {
			echo_debug('Resetting connection count', DEBUG_SERVER);
			$this->reset_connection_limits();
			$this->conn_check_time = 0;
		}
	}

	function __destruct() 
	{
		 /* nothing yet */ 
    }
    function has_nntp_task()
    {
        return $this->free_nntp_slots == $this->max_total_nntp_threads;
    } 
	function has_equal(action $a)
	{
		return $this->queue->has_equal($a) || $this->threads->has_equal($a); 
	}

	// schedule wrappers
	function get_first_timeout() 
	{ 
		return $this->schedule->get_first_timeout(); 
	}
	function add_schedule(DatabaseConnection $db, job $job)
	{
		return $this->schedule->add($db, $job);
	}
	function schedule_size()
	{
		return $this->schedule->size();
	}
	function get_jobs() 
	{ 
		return $this->schedule->get_jobs();
	}
	function recur_schedule(DatabaseConnection $db, job $job) 
	{
		$action = $job->get_action();
		$rec = $job->get_recurrence();
		if ($rec > 0) {
			$new_action = new action(NULL, NULL, NULL);
			$new_action->copy($action);
			$time = $job->get_time();
			$new_job = new job($new_action, $time + $rec, $rec);
			$this->add_schedule($db, $new_job);
		}
	}
	function nntp_slots_available()
	{
		return ($this->free_nntp_slots > 0 && $this->free_total_slots > 0) ? TRUE : FALSE; 
	}
	function slots_available()
	{
		return ($this->free_total_slots > 0) ? TRUE : FALSE; 
	}
	function unschedule(DatabaseConnection $db, $username, $arg) 
	{ 
		return $this->schedule->unschedule($db, $username, $arg); 
	}
	function unschedule_all(DatabaseConnection $db, $username) 
	{
		return $this->schedule->unschedule_all($db, $username);
	}
	function unschedule_cmd(DatabaseConnection $db, $username, $cmd, $arg)
       	{
		return $this->schedule->unschedule_cmd($db, $username, $cmd, $arg);
	}
    function get_first_ready_job(DatabaseConnection $db)
    {
        return $this->schedule->get_first_ready_job($db);
    }
    function recreate_post_command(DatabaseConnection $db, action $item, $pause = FALSE, $reset_servers=FALSE)
	{
		assert(is_bool($pause) && is_bool($reset_servers));
		$new_item = new action(NULL, NULL, NULL);
		$new_item->copy($item);
		$new_item->set_command(COMMAND_START_POST);
		if ($reset_servers === TRUE)
			$new_item->reset_tried_servers();
		if ($pause === TRUE)
			$new_item->pause($db, TRUE, 'root');
		$this->queue_push($db, $new_item, FALSE);
		return $new_item->get_id();
	}

	function recreate_download_command(DatabaseConnection $db, action $item, $pause = FALSE, $reset_servers=FALSE)
	{
		assert(is_bool($pause) && is_bool($reset_servers));
		$new_item = new action(NULL, NULL, NULL);
		$new_item->copy($item);
		$new_item->set_command(COMMAND_DOWNLOAD);
		if ($reset_servers === TRUE)
			$new_item->reset_tried_servers();
		if ($pause === TRUE)
			$new_item->pause($db, TRUE, 'root');
		$this->queue_push($db, $new_item, FALSE);
		return $new_item->get_id();
	}
	function unused_servers_available(array $tried_servers)
	{
		return $this->servers->unused_servers_available($tried_servers);
    }
	function reschedule_download(DatabaseConnection $db, action &$item, $server_id)
	{// cleanup this one XXX
		echo_debug_function(DEBUG_MAIN, __FUNCTION__);
		assert(is_numeric($server_id));
		if (!$this->threads->has_equal($item)) {
			// this is the last running one
			// diminish amount of threads on server
			$status = QUEUE_CANCELLED;

			$this->servers->dec_free_slots($server_id);
			if ($this->conn_check_time == 0) // set the connection retry timeout if not set
				$this->conn_check_time = time() + CONNECT_CHECK_TIME;
			$this->queue->delete_cmd($db, $item->get_command(), $item->get_args(), 'root', TRUE); // remove all equal from queue
			if ($this->servers->unused_servers_available($item->get_all_failed_servers()) !== FALSE) {// we can try other servers
				$this->recreate_download_command($db, $item, FALSE, FALSE);
			} else if ($item->get_preview()) { // it's a preview that can't start on any server... need to preempt a task
				try {
					$username = $item->get_username();
					$rv = $this->queue_push($db, $item, FALSE, QUEUE_TOP); 
					if ($rv === FALSE){
						update_dlinfo_status($db, DOWNLOAD_PAUSED, $item->get_args());
					} else {
						$this->preempt($db, $item, $username);	
					}
				} catch (exception $e) {
					echo_debug_trace($e, DEBUG_MAIN);
					write_log("Fatal error: {$e->getMessage()}", LOG_ERR);
				}
			} else {// all servers tried, queue it paused
				echo_debug('Requeueing paused', DEBUG_SERVER);
				$new_id = $this->recreate_download_command($db, $item, TRUE, TRUE);
				update_dlinfo_status($db, DOWNLOAD_PAUSED, $item->get_args());
				$item_unpause = new action (COMMAND_CONTINUE, $new_id, $item->get_username(), TRUE, $item->get_priority());
				$job = new job($item_unpause, time() + get_timeout($item) * 60, NULL); //try again in 60 secs
				$this->schedule->add ($db, $job);
			}
		} else { // we simple pause the item for a while
			$this->reschedule ($db, $item, $server_id);
		}
	}

	function reschedule_post(DatabaseConnection $db, action &$item, $server_id)
	{// cleanup this one XXX
		echo_debug_function(DEBUG_MAIN, __FUNCTION__);
		assert(is_numeric($server_id));
		if (!$this->threads->has_equal($item)) {
			// this is the last running one
			// diminish amount of threads on server
			$status = QUEUE_CANCELLED;

			$this->servers->dec_free_slots($server_id);
			if ($this->conn_check_time == 0) // set the connection retry timeout if not set
				$this->conn_check_time = time() + CONNECT_CHECK_TIME;
			$this->queue->delete_cmd($db, $item->get_command(), $item->get_args(), 'root', TRUE); // remove all equal from queue
			if ($this->servers->unused_servers_available($item->get_all_failed_servers()) !== FALSE) {// we can try other servers
				$this->recreate_post_command($db, $item, FALSE, FALSE);
			} else {// all servers tried, queue it paused
				echo_debug('Requeueing paused', DEBUG_SERVER);
				$new_id = $this->recreate_post_command($db, $item, TRUE, TRUE); //xxx
				update_postinfo_status($db, POST_PAUSED, $item->get_args());
				$item_unpause = new action (COMMAND_CONTINUE, $new_id, $item->get_username(), TRUE, $item->get_priority());
				$job = new job($item_unpause, time() + get_timeout($item) * 60, NULL); //try again in 60 secs
				$this->schedule->add ($db, $job);
			}
		} else { // we simple pause the item for a while
			$this->reschedule ($db, $item, $server_id);
		}
	}
	function reschedule(DatabaseConnection$db, action &$item, $server_id)
	{// cleanup this one XXX
		// diminish amount of threads on server
		assert(is_numeric($server_id));
		$this->servers->dec_free_slots($server_id);
		if ($this->conn_check_time == 0)
			$this->conn_check_time = time() + CONNECT_CHECK_TIME;
		if ($server_id !== FALSE && $this->servers->has_free_slot($server_id)){
			echo_debug('Requeueing with one connection less', DEBUG_SERVER);
			$this->queue_push($db, $item, FALSE);
		} else {
			echo_debug('Requeueing paused', DEBUG_SERVER);
			$item->pause($db, TRUE, 'root');
			$this->queue_push($db, $item, FALSE);
			$item_unpause = new action (COMMAND_CONTINUE, $item->get_id(), $item->get_username(),TRUE, $item->get_priority());
			$job = new job($item_unpause, time() +  get_timeout($item) * 60, NULL); //try again in 60 secs
			$this->schedule->add ($db, $job);
		}
	}
	function reload_scheduled_jobs(DatabaseConnection $db)
	{
		echo_debug_function(DEBUG_MAIN, __FUNCTION__);
		$query = 'SELECT * FROM schedule WHERE 1=1';
		$res = $db->execute_query($query);
		if ($res === FALSE)
			return;
		foreach ($res as $sched) {
			$command = $sched['command'];
			$c = explode (" ", $command, 2);
			$cmd = $c[0];
			$arg = (isset ($c[1]))? $c[1] : '';
			$username = $sched['username'];
			$stime = $sched['at_time'];
			$repeat = $sched['interval'];
			$id = $sched['id'];
			$ctime = time();
			if ($repeat > 0) {
				while (($ctime + min(60, $repeat)) > $stime) 
					$stime += $repeat;
			} else {
				if ($stime < $ctime)
					$stime = $ctime;
				$repeat = NULL;
			}

			$item = new action ($cmd, $arg, $username);
			$this->schedule->add($db, new job($item, $stime, $repeat));
			$q = "DELETE FROM schedule WHERE \"id\"=$id";
			$res = $db->execute_query($q);
		}
	}
	//usenet_wrappers
	function reset_connection_limits() 
	{ 
		$this->servers->reset_connection_limits(); 
	}
	function set_update_server ($srv) 
	{
		assert(is_numeric($srv));
		$this->servers->set_update_server($srv);
	}
	function get_update_server() 
	{
		return $this->servers->get_update_server();
	}

	function reload_server(DatabaseConnection $db, $id)
	{
		assert(is_numeric($id));
		$srv = get_usenet_server($db, $id, FALSE);
		$this->servers->update_server($srv, $id);
	}
	function load_server(DatabaseConnection$db, $id)
	{
		assert(is_numeric($id));
		$s = get_usenet_server($db, $id, FALSE);
		$us = new usenet_server($s['id'], $s['threads'], $s['port'], $s['hostname'], $s['connection'], $s['username'], $s['password'], $s['priority'], $s['posting']);
		$this->servers->add_server($us);
	}
	function load_servers(DatabaseConnection $db, test_result_list &$test_results)
	{
		$srv = get_all_usenet_servers($db, FALSE);
		$cnt = 0;
		foreach($srv as $s) {
			$us = new usenet_server($s['id'], $s['threads'], $s['port'], $s['hostname'], $s['connection'], $s['username'], $s['password'], $s['priority'], $s['posting']);
			$cnt += $this->add_server($us);
		}
        if ($cnt == 0) 
            throw new exception('Enable at least one server in admin/usenet servers', ERR_NO_ACTIVE_SERVER);

        $update_server = get_config($db, 'preferred_server');
        if ($update_server == 0)
            throw new exception('Please set update server in admin/usenet servers', ERR_NO_ACTIVE_SERVER);
        try {
            $this->servers->set_update_server($update_server);
        } catch (exception $e) {
            if ($e->getcode() == ERR_NO_SUCH_SERVER)
                throw new exception('Please set a valid update server in admin/usenet servers', ERR_NO_ACTIVE_SERVER);
            else
                throw $e;
        }
        $this->servers->find_max_connections($db, $test_results);
    }
    function get_max_threads($server_id) 
    { 
        assert(is_numeric($server_id));
        return $this->servers->get_max_threads($server_id); 
    } 
    function find_preferred_server(DatabaseConnection $db) 
    { 
        try {
            $id = $this->servers->get_update_server();
        } catch (exception $e) {
            write_log ('No primary server set', LOG_ERR);
            throw $e;
        }
        return $id;
    }
    function add_server(usenet_server $server) 
    {
        return $this->servers->add_server($server); 
    }
    function delete_server($id) 
    {
        assert(is_numeric($id));
        $this->servers->delete_server($id); 
    } 
    function enable_posting($server_id) 
    {
        assert(is_numeric($server_id));
        return $this->servers->enable_posting($server_id); 
    }
    function disable_posting($server_id) 
    {
        assert(is_numeric($server_id));
        return $this->servers->disable_posting($server_id); 
    }
 
    function enable_server($server_id, $prio) 
    {
        assert(is_numeric($server_id) && is_numeric($prio));
        return $this->servers->enable_server($server_id, $prio); 
    }
    function disable_server($server_id) 
    {
        assert(is_numeric($server_id));
        return $this->servers->disable_server($server_id); 
    }
    function get_servers(DatabaseConnection $db)
    { 
        return $this->servers->get_servers($db) ;
    } 
    function find_free_slot(array $already_used_servers=array(), $posting) 
    { 
        return $this->servers->find_free_slot($already_used_servers, $posting); 
    }
    // queue wrappers
    function pause(DatabaseConnection $db, $id, $pause, $username)
    {
        assert(is_numeric($id) && is_bool($pause));
        try {
            $rv = $this->queue->pause($db, $id, $pause, $username);
        } catch (exception $e) {
            $rv = FALSE;
        }
        if ($pause === TRUE) { // can't restart running threads so only check threads if we are pausing, not continuing
            try {
                $pid = $this->threads->get_pid($id);
                $thread =& $this->threads->get_thread($pid);
                if (!$thread->get_action()->match_username($username))
                    throw new exception ('Not allowed', ERR_ACCESS_DENIED);
                urdd_kill($pid, SIGTERM);

                $thread->set_status(DOWNLOAD_PAUSED);
                $item = $thread->get_action();
                $cmd = $item->get_command();
                if (compare_command($cmd, COMMAND_DOWNLOAD) || compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
                    $dlid = $item->get_args();
                    update_dlinfo_status($db, DOWNLOAD_PAUSED, $dlid);
                } else if (compare_command($cmd, COMMAND_START_POST) || compare_command($cmd, COMMAND_POST_ACTION) || compare_command($cmd, COMMAND_POST)) { 
                    $dlid = $item->get_args();
                    update_postinfo_status($db, POST_PAUSED, $dlid);
                }
                $item = $thread->get_action(); // delete from the list will be done by the reap function
                $item->pause($db, $pause, $username);
                $this->queue_push($db, $item, FALSE); // if it is paused we simply put it back on the queue
                return TRUE;
            } catch (exception $e){
                return $rv;
            }
        }
    }
    function move(DatabaseConnection $db, $cmd, $arg, $username, $direction) 
    {
        $this->queue->move($db, $cmd, $arg, $username, $direction);
    }
    function pause_cmd(DatabaseConnection $db, $cmd, $arg, $do_pause, $username)// $do_pause == true then pause, else continue (unpause)
    {
        assert(is_bool($do_pause));
        echo_debug_function(DEBUG_SERVER, __FUNCTION__);
        try {
            $rv = $this->queue->pause_cmd($db, $cmd, $arg, $do_pause, $username);
        } catch (exception $e) { 
            $rv = FALSE;
        } 
        if ($do_pause === TRUE) { // can't restart running threads so only check threads if we are pausing, not continuing
            $pids = $this->threads->get_pid_cmd($cmd, $arg);
            foreach ($pids as $pid) {
                $thread =& $this->threads->get_thread($pid);
                if (!$thread->get_action()->match_username($username))
                    continue;
                urdd_kill($pid, SIGTERM);
                $thread->set_status(DOWNLOAD_PAUSED);
				$item = $thread->get_action(); // delete from the list will be done by the reap function
				$cmd = $item->get_command();
				if (compare_command($cmd, COMMAND_DOWNLOAD) || compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
					$dlid = $item->get_args();
					update_dlinfo_status($db, DOWNLOAD_PAUSED, $dlid);
                } else if (compare_command($cmd, COMMAND_START_POST) ||compare_command($cmd, COMMAND_POST_ACTION) ||compare_command($cmd, COMMAND_POST)) { 
					$dlid = $item->get_args();
					update_postinfo_status($db, POST_PAUSED, $dlid);
				}

                $item->pause($db, $do_pause, $username);
				$this->queue_push($db, $item, FALSE); // if it is paused we simply put it back on the queue
				$rv = TRUE;
			}
		}
		return $rv;
	}
	function pause_all(DatabaseConnection$db, $do_pause, $username)
	{
		assert(is_bool($do_pause));
		$this->queue->pause_all($db, $do_pause, $username);
		if ($do_pause === TRUE) {
			$pids = $this->threads->get_all_pids($username);
			foreach($pids as $pid) {
				$thread =& $this->threads->get_thread($pid);
				if ($thread->get_action()->match_username($username)) {
					$thread->set_status(DOWNLOAD_PAUSED);
					$item = $thread->get_action(); // delete from the list will be done by the reap function
                    urdd_kill($pid, SIGTERM);

                    $item->pause($db, $do_pause, $username); // we set it to pause
                    $cmd = $item->get_command();
					if (compare_command($cmd, COMMAND_DOWNLOAD) ||compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
						$dlid = $item->get_args();
						update_dlinfo_status($db, DOWNLOAD_PAUSED, $dlid);
					} else if (compare_command($cmd, COMMAND_START_POST) ||compare_command($cmd, COMMAND_POST_ACTION) ||compare_command($cmd, COMMAND_POST)) { 
						$dlid = $item->get_args();
						update_postinfo_status($db, POST_PAUSED, $dlid);
					}
					$this->queue_push($db, $item, FALSE); // and reschedule it
				}
			}
		}

	}
	function queue_push(DatabaseConnection $db, action $item, $increase_counter=TRUE, $position=QUEUE_BOTTOM, $priority=NULL)
	{
		assert(is_bool($increase_counter));
		echo_debug('Queueing ' . $item->get_command() . ' ' . $item->get_args(), DEBUG_SERVER);
		switch ($position) {
		case QUEUE_TOP:
			return $this->queue->push_top($item, $db, $increase_counter);
			break;
		case QUEUE_BOTTOM:
			return $this->queue->push($item, $db, $increase_counter, $priority);
			break;
		default:
			throw new exception ('Don\'t understand queue position', ERR_QUEUE_FAILED);

		}
	}
    function queue_size() 
    { 
        return $this->queue->size(); 
    }
    function move_top(DatabaseConnection $db, $index, $username) 
    { 
        assert(is_numeric($index));
        return $this->queue->move_top($db, $index, $username); 
    }
	function queue_delete(DatabaseConnection $db, $action_id, $username, $delete_db = FALSE) 
	{ 
		assert(is_numeric($action_id) && is_bool($delete_db));
		echo_debug_function(DEBUG_MAIN, __FUNCTION__);
		return $this->queue->delete($db, $action_id, $username, $delete_db); 
    }
	protected function update_dl_status(DatabaseConnection $db, $q_ids, $dl_status, $post_status)
	{
		foreach($q_ids as $qid) {
			$item = $this->queue->get_queue_item($qid);
			if ($item === NULL) continue; // need warning here?

			$cmd = $item->get_command();
			if (compare_command($cmd, COMMAND_POST) || compare_command($cmd, COMMAND_POST_ACTION) || compare_command($cmd, COMMAND_POST)) { 
				$dlid = $item->get_args();
				update_postinfo_status($db, $post_status, $dlid);
            } else if (compare_command($cmd, COMMAND_DOWNLOAD) || compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
				$dlid = $item->get_args();
				update_dlinfo_status($db, $dl_status, $dlid);
			}
		}
	}
	function delete_cmd(DatabaseConnection $db, $username, $cmd, $arg, $delete_db = FALSE)
	{ 
		assert(is_bool($delete_db));
		// XXX fix so that queue returns the queue_items and then run a delete on all of them, and updating the db dl status too
		//		$rv = $this->queue->delete_cmd($db, $cmd, $arg, $username, $delete_db);

		$q_ids = $this->queue->get_ids_cmd($cmd, $arg, $username);
		if ($delete_db === TRUE ) 
            $this->update_dl_status($db, $q_ids, DOWNLOAD_CANCELLED, POST_CANCELLED);
       
		$rv = $this->queue->delete_ids($db, $q_ids, $username, $delete_db);
		$rv = $rv > 0;
		$pids = $this->threads->get_pid_cmd($cmd, $arg);
		foreach ($pids as $pid) {
			$thread =& $this->threads->get_thread($pid);
			if (!$thread->get_action()->match_username($username))
				throw new exception ('Not allowed', ERR_ACCESS_DENIED);
            urdd_kill($pid, SIGTERM);

			if ($delete_db === TRUE) {
				$item = $thread->get_action();
				$thread->set_status(DOWNLOAD_CANCELLED);
				if (compare_command($cmd, COMMAND_DOWNLOAD) || compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
					$dlid = $item->get_args();
					update_dlinfo_status($db, DOWNLOAD_CANCELLED, $dlid);
				} else if (compare_command($cmd, COMMAND_START_POST) || compare_command($cmd, COMMAND_POST_ACTION) || compare_command($cmd, COMMAND_POST)) { 
					$dlid = $item->get_args();
					update_postinfo_status($db, POST_CANCELLED, $dlid);
				}
				update_queue_status($db, $thread->get_action()->get_dbid(), QUEUE_CANCELLED);
			}
			$rv = TRUE;
		}
		return $rv;
	}
	function delete_all(DatabaseConnection $db, $username, $delete_db=FALSE)
	{
		assert(is_bool($delete_db));
		$q_ids = $this->queue->get_ids_all($username);
		if ($delete_db === TRUE )
			$this->update_dl_status($db, $q_ids, DOWNLOAD_CANCELLED, POST_CANCELLED);
		$rv = $this->queue->delete_ids($db, $q_ids, $username, $delete_db);
		$pids = $this->threads->get_all_pids($username);
		foreach($pids as $pid) {
			$thread = &$this->threads->get_thread($pid);
			if ($delete_db === TRUE) {
				$item = $thread->get_action();
				$cmd = $item->get_command();
				if (compare_command($cmd, COMMAND_DOWNLOAD) ||compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
					$dlid = $item->get_args();
					update_dlinfo_status($db, DOWNLOAD_CANCELLED, $dlid);
				} else if (compare_command($cmd, COMMAND_START_POST) || compare_command($cmd, COMMAND_POST_ACTION) || compare_command($cmd, COMMAND_POST)) { 
					$dlid = $item->get_args();
					update_postinfo_status($db, POST_CANCELLED, $dlid);
				}
				update_queue_status($db, $item->get_dbid(), QUEUE_CANCELLED);
				$thread->set_status(DOWNLOAD_CANCELLED);
			} else
				$thread->set_status(DOWNLOAD_SHUTDOWN);

            urdd_kill($pid, SIGTERM);

			// updating database is done by reap function
		}// XXX fix so that queue returns the queue_items and then run a delete on all of them, and updating the db dl status too
        //		return $this->queue->delete_all($db, $username, $delete_db);
        $this->restore_server_settings();
        return TRUE; // todo
	}
	function delete(DatabaseConnection $db, $action_id, $username, $delete_db = FALSE)
	{
		assert(is_numeric($action_id) && is_bool($delete_db));
		// XXX fix so that queue returns the queue_items and then run a delete on all of them, and updating the db dl status too
		//		$rv = $this->queue->delete($db, $action_id, $username, $delete_db);
		$q_ids = $this->queue->get_ids_action($action_id, $username);
		if ($delete_db === TRUE )
			$this->update_dl_status($db, $q_ids, DOWNLOAD_CANCELLED, POST_CANCELLED);
		$rv = $this->queue->delete_ids($db, $q_ids, $username, $delete_db);
		$rv = $rv > 0;

		if ($rv === FALSE) {
			$pid = $this->threads->get_pid($action_id);
			if ($pid !== FALSE) {
				$thread = &$this->threads->get_thread($pid);
				$item = $thread->get_action();
				if (!$item->match_username($username))
					throw new exception ('Not allowed', ERR_ACCESS_DENIED);
				if ($delete_db === TRUE) {
					$cmd = $item->get_command();
					if (compare_command($cmd, COMMAND_DOWNLOAD) || compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
						$dlid = $item->get_args();
						update_dlinfo_status($db, DOWNLOAD_CANCELLED, $dlid);
					} else if (compare_command($cmd, COMMAND_START_POST) || compare_command($cmd, COMMAND_POST_ACTION) || compare_command($cmd, COMMAND_POST)) { 
                        $dlid = $item->get_args();
                        update_postinfo_status($db, POST_CANCELLED, $dlid);
                    }

					$thread->set_status(DOWNLOAD_CANCELLED);
					update_queue_status($db, $thread->get_action()->get_dbid(), QUEUE_CANCELLED);
				}
                urdd_kill($pid, SIGTERM);
				$rv = TRUE;
			}
		}
		return $rv;
	}
	function queue_set_priority(DatabaseConnection $db, $action_id, $username, $priority) 
	{ 
		assert(is_numeric($action_id) && is_numeric($priority));
		return $this->queue->set_priority($db, $action_id, $username, $priority); 
	}
	function get_queue() 
	{
		return $this->queue->get_queue();
	} 

	function get_first_runnable_on_queue(DatabaseConnection $db)
	{ // returns TRUE if the queue is empty, FALSE if no item is runnable, otherwise it returns the runnable item
		try {
			if ($this->queue->is_empty()) // we have nothing to do
				return TRUE;
			$item = $this->queue->get_preview_action(); // see if there are any preview tasks queued
			if ($item !== FALSE) { // there is one preview
				$ready = TRUE;
				if($item->get_preferred_server() == 0 && in_array($item->get_command_code(), array(COMMAND_DOWNLOAD_ACTION, COMMAND_DOWNLOAD)) && 
					$this->free_total_slots > 0 && $this->free_nntp_slots > 0) { 
						// so it must not have a server set yet and it must be a download action and there must be a total slot available
						$srv = $this->servers->find_free_slot($item->get_all_failed_servers(), FALSE); // find a server with a slot available
						if($srv !== FALSE && $srv != 0 && $this->free_total_slots > 0) { 
							// so there must be a free server, and there must be a slot available to set the preferred server
							$item->set_preferred_server($srv); // set the prefered server
							echo_debug('Found a server ' . $srv ,  DEBUG_SERVER);
							$ready = TRUE;
						} else {
                            echo_debug('Could not find a server', DEBUG_SERVER);
                        }
					} else
						$ready = TRUE;

				if ($ready === TRUE) {
                    echo_debug('Found a preview thread... should always start '. $item->get_preferred_server(), DEBUG_SERVER);
                    return $item;
                }
			}

			if ($this->free_total_slots <= 0) {
				return TRUE; // no slot available....
			} else if ($this->free_nntp_slots <= 0)  { // no nntp slot available but maybe there is a non-nntp item on the queue
				$item = $this->queue->top(FALSE, array(), $this->free_db_intensive_slots > 0 ? TRUE : FALSE ); 
				if ($item === FALSE)
					return TRUE;

				if ($item->db_intensive() === TRUE) {
					if ($this->free_db_intensive_slots > 0) {
						$this->free_db_intensive_slots--;
						return $item;
					} else {
						throw new exception ('Something has gone terribly wrong', ERR_INTERNAL_ERROR);
						return FALSE; // something is wrong !!!
					}
				} else
					return $item;
			} else {
				$not_these = array() ; // the ones we already tried
				while (1) {
					$item = $this->queue->top(TRUE, $not_these, $this->free_db_intensive_slots > 0 ? TRUE : FALSE); // get an nntp item from the queue or if non found a nonnntp item
					if ($item === FALSE) // there are no more items that can run
						return TRUE;
					if ($item->db_intensive() === TRUE) {
						if ($this->free_db_intensive_slots > 0) {
							$this->free_db_intensive_slots--;
							return $item;
						} else  {
							throw new exception ('Something has gone terribly wrong', ERR_INTERNAL_ERROR);
							return FALSE; // something is wrong !!!
						}
					}
					if ($item->need_nntp() === FALSE) // non-nntp items can always run if there is a slot available
						return $item;
					if ($item->primary_nntp()) {
						$server_id = $this->get_update_server();
						$item->set_preferred_server($server_id);
					} else 
						$server_id = $item->get_preferred_server();
					if ($server_id == 0) {// if the preferred server is not set, it can probably run
						$srv = $this->servers->find_free_slot($item->get_all_failed_servers(), $item->need_posting());
						if ($srv !== FALSE) {
							$item->set_preferred_server($srv);
							// check if a free slot is available
							return $item;
						}
					} else if ($this->servers->has_free_slot($server_id)) { // check if the preferred server is available
						return $item;
					}
					$not_these[] = $item->get_id(); // not available lets not try this queued item again through this iteration
				}	
			} 
		} catch (exception $e) {
			// something went wrong,... but what
			echo_debug('o-oh an error... ' . $e->getMessage() . ' ' . $e->getCode(), DEBUG_SERVER);
			return FALSE;
		}
		return FALSE; // safety catch: shouldn't happen tho
	}

	// threads wrappers
	function add_thread(DatabaseConnection $db, thread $thread, $server_id=FALSE)
	{
		assert(is_numeric($server_id) || $server_id === FALSE);
		$action = $thread->get_action();
		$pid = $thread->get_pid();
		$need_nntp = $action->need_nntp();
		if ($this->free_total_slots <= 0) {// check if there is a slot available... 
			$this->threads->add_dummy_thread($pid);
			throw new exception ('no slot available', ERR_NO_SLOT_AVAILABLE);
		}
        if ($need_nntp === TRUE) { // we are an nntp item check if there is an nntp slot available on any of the servers
            if ($this->free_nntp_slots <= 0) {
				$this->threads->add_dummy_thread($pid);
				throw new exception ('no nntp slot available 2', ERR_NO_NNTPSLOT_AVAILABLE);
			}

			$server_id = $action->get_preferred_server();
			if ($server_id == 0) 
				$server_id = FALSE;
			if ($server_id === FALSE)
				$server_id = $this->servers->find_free_slot($action->get_all_failed_servers(), $action->need_posting());
			if ($server_id === FALSE) {
				$this->threads->add_dummy_thread($pid);
				throw new exception ('no nntp slot available 1', ERR_NO_NNTPSLOT_AVAILABLE);
			}
            $this->threads->add_thread($db, $thread, $server_id);
            $this->servers->add_thread($server_id);
			$this->free_nntp_slots--;
            $this->free_total_slots--;
		} else {
			$this->threads->add_thread($db, $thread, 0);
			$this->free_total_slots--;
		}
	}
	function delete_thread(DatabaseConnection $db, $pid, $store)
	{
		assert(is_numeric($pid) && is_bool($store));
		list ($item, $server_id, $status) = $this->threads->delete_thread($db, $pid, $store);
		$this->free_total_slots++;
		if ($item->need_nntp()) {
            $this->free_nntp_slots++;
            $this->servers->delete_thread($server_id);
		}
		if ($item->db_intensive())
			$this->free_db_intensive_slots++;
		return array($item, $server_id, $status);
	}
	function threads_size() 
	{
		return $this->threads->size();
	}
	function get_threads()
	{ 
		return $this->threads->get_threads();
	} 
	function preempt(DatabaseConnection $db, action $item, $username)
	{
		echo_debug_function(DEBUG_MAIN, __FUNCTION__);
		$pids = $this->threads->get_all_pids($username);
		$old_id = FALSE;
		foreach ($pids as $item_pid) {
			$t = $this->threads->get_thread($item_pid);
			$item2 = $t->get_action();
            if ($item2->get_preview()) 
                continue; // preview threads won't get preempted
			$old_id = $item2->get_id();
            if (($item2->need_nntp() == $item->need_nntp()) 
                && !in_array($item2->get_active_server(), $item->get_failed_servers()) 
                && !in_array($item2->get_active_server(), $item->get_tried_servers())) // this one will do
				break;
			// otherwise we take the last one... this means no nntp threads were in use, but all were occupied by non nntp threads 
			// or we run a non nntp thread and no nonnntp are running ;-)
		}
		if ($old_id === FALSE) {
			$this->delete($db, $item->get_id(), $username, TRUE); 
			throw new exception ('Cannot preempt a task');
		}
		try {
			$this->stop($db, $old_id, $username);
		} catch (exception $e) { 
			throw new exception ('ID not found ' . $e->getMessage());
		}
		$this->queue->move_top($db, $item->get_id(), 'root');
		return $item2->get_active_server();
	}
	function stop_all(DatabaseConnection $db, $username)
	{
		$pids = $this->threads->get_all_pids($username);
		foreach($pids as $pid) {
			$thread =& $this->threads->get_thread($pid);
			if ($thread->get_action()->match_username($username)) {
				$thread->set_status(DOWNLOAD_STOPPED);
				$item = $thread->get_action(); // delete from the list will be done by the reap function
                urdd_kill($pid, SIGTERM);

                //				pcntl_sigwaitinfo(array(SIGCHLD)); add for php 0.5.3
				$this->queue_push($db, $item, TRUE); // and reschedule it
			}
		}
	}
	function stop(DatabaseConnection $db, $id, $username) 
	{
		assert(is_numeric($id));
		$pid = $this->threads->get_pid($id);

		$thread =& $this->threads->get_thread($pid);
		if (!$thread->get_action()->match_username($username)) 
			throw new exception('Not allowed', ERR_ACCESS_DENIED);
		$thread->set_status(DOWNLOAD_STOPPED);
		$item = $thread->get_action(); // delete from the list will be done by the reap function
		$item->clear_tried_servers(array($item->get_active_server()));
        $this->queue_push($db, $item, FALSE, QUEUE_BOTTOM, 3); // and reschedule it to the top of the queue, but after previews

        urdd_kill($pid, SIGTERM);
		return TRUE;

	}
	function test_servers(DatabaseConnection $db,test_result_list &$test_results)
	{
		$this->servers->test_servers($db, $test_results);
	}
	function find_servers(DatabaseConnection $db, test_result_list &$test_results, $dbid, $type, $update_config)
	{
		$this->servers->find_servers($db, $test_results, $dbid, $type, $update_config);
    }
    function restore_server_settings()
    {
        $this->free_nntp_slots = $this->max_total_nntp_threads;
        $this->free_total_slots = $this->max_total_threads ;
        $this->free_db_intensive_slots = $this->max_db_intensive_threads;

        $this->servers->restore_server_settings();
    }

} // server data


?>
